/////////////////////////////////////////////////////////////////////////////
// Original code from libhdfs3. Copyright (c) 2013 - 2014, Pivotal Inc.
// All rights reserved. Author: Zhanwei Wang
/////////////////////////////////////////////////////////////////////////////
//  Modifications by Kumo Inc.
// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//



#pragma once


#include <memory>
#include <kmhdfs/rpc/rpc_auth.h>
#include <kmhdfs/rpc/rpc_call.h>
#include <kmhdfs/rpc/rpc_channel.h>
#include <kmhdfs/rpc/rpc_channel_key.h>
#include <kmhdfs/rpc/rpc_config.h>
#include <kmhdfs/rpc/rpc_protocol_info.h>
#include <kmhdfs/rpc/rpc_server_info.h>
#include <kmhdfs/common/thread.h>

#include <vector>

#ifdef MOCK
#include <tests/mock/test_rpcchannel_stub.h>
#endif

namespace Hdfs {
    namespace Internal {
        class RpcClient {
        public:
            /**
             * Destroy an RpcClient instance.
             */
            virtual ~RpcClient() {
            }

            /**
             * Get a RPC channel, create a new one if necessary.
             * @param auth Authentication information used to setup RPC connection.
             * @param protocol The RPC protocol used in this call.
             * @param server Remote server information.
             * @param conf RPC connection configuration.
             * @param once If true, the RPC channel will not be reused.
             */
            virtual RpcChannel &getChannel(const RpcAuth &auth,
                                           const RpcProtocolInfo &protocol, const RpcServerInfo &server,
                                           const RpcConfig &conf) = 0;

            /**
             * Check the RpcClient is still running.
             * @return true if the RpcClient is still running.
             */
            virtual bool isRunning() = 0;

            virtual std::string getClientId() const = 0;

            virtual int32_t getCallId() = 0;

        public:
            static RpcClient &getClient();

            static void createSinglten();

        private:
            static once_flag once;
            static std::shared_ptr<RpcClient> client;
        };

        class RpcClientImpl : public RpcClient {
        public:
            /**
             * Construct a RpcClient.
             */
            RpcClientImpl();

            /**
             * Destroy an RpcClient instance.
             */
            ~RpcClientImpl();

            /**
             * Get a RPC channel, create a new one if necessary.
             * @param auth Authentication information used to setup RPC connection.
             * @param protocol The RPC protocol used in this call.
             * @param server Remote server information.
             * @param conf RPC connection configuration.
             * @param once If true, the RPC channel will not be reused.
             */
            RpcChannel &getChannel(const RpcAuth &auth,
                                   const RpcProtocolInfo &protocol, const RpcServerInfo &server,
                                   const RpcConfig &conf);

            /**
             * Close the RPC channel.
             */
            void close();

            /**
             * Check the RpcClient is still running.
             * @return true if the RpcClient is still running.
             */
            bool isRunning();

            std::string getClientId() const {
                return clientId;
            }

            int32_t getCallId() {
                static mutex mutid;
                lock_guard<mutex> lock(mutid);
                ++count;
                count = count < std::numeric_limits<int32_t>::max() ? count : 0;
                return count;
            }

        private:
            std::shared_ptr<RpcChannel> createChannelInternal(
                const RpcChannelKey &key);

            void clean();

        private:
            std::atomic<bool> cleaning;
            std::atomic<bool> running;
            condition_variable cond;
            int64_t count;
            mutex mut;
            std::string clientId;
            thread cleaner;
            std::unordered_map<RpcChannelKey, std::shared_ptr<RpcChannel> > allChannels;

#ifdef MOCK
        private:
            /*
             * for test
             */
            Hdfs::Mock::TestRpcChannelStub *stub;
#endif
        };
    }
}

